package com.alibaba.nacos.naming.remote.rpc.handler;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import com.alibaba.nacos.api.naming.remote.request.SubscribeServiceRequest;
import com.alibaba.nacos.api.naming.remote.response.SubscribeServiceResponse;
import com.alibaba.nacos.api.remote.request.RequestMeta;
import com.alibaba.nacos.api.remote.response.ResponseCode;
import com.alibaba.nacos.core.remote.RequestHandler;
import com.alibaba.nacos.naming.core.v2.index.ServiceStorage;
import com.alibaba.nacos.naming.core.v2.metadata.NamingMetadataManager;
import com.alibaba.nacos.naming.core.v2.metadata.ServiceMetadata;
import com.alibaba.nacos.naming.core.v2.pojo.Service;
import com.alibaba.nacos.naming.core.v2.service.ClientOperationService;
import com.alibaba.nacos.naming.pojo.Subscriber;
import com.alibaba.nacos.naming.utils.ServiceUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class SubscribeServiceRequestHandler
        extends RequestHandler<SubscribeServiceRequest, SubscribeServiceResponse> {

    @Autowired
    private ClientOperationService clientOperationService;
    @Autowired
    private ServiceStorage serviceStorage;
    @Autowired
    private NamingMetadataManager namingMetadataManager;

    @Override
    public SubscribeServiceResponse handle(SubscribeServiceRequest request,
                                           RequestMeta meta) throws NacosException {
        Service service = Service.newService(request.getNamespace(), request.getGroupName(), request.getServiceName(), true);
        Subscriber subscriber = new Subscriber().setAddrStr(meta.getClientIp())
                .setAgent(meta.getClientVersion())
                .setApp("")
                .setIp(meta.getClientIp())
                .setNamespaceId(service.getNamespace())
                .setServiceName(service.getGroupedServiceName())
                .setPort(0)
                .setCluster(request.getClusters());

        if (request.isSubscribe()) {
            clientOperationService.subscribeService(service, subscriber, meta.getConnectId());
        } else {
            clientOperationService.unsubscribeService(service, subscriber, meta.getConnectId());
        }

        ServiceInfo serviceInfo = serviceStorage.getData(service);
        ServiceMetadata serviceMetadata = namingMetadataManager.getServiceMetadata(service);
        ServiceInfo result = ServiceUtil.selectInstancesWithHealthyProtection(serviceInfo, serviceMetadata, subscriber.getCluster(), false, true, subscriber.getIp());
        return new SubscribeServiceResponse(ResponseCode.SUCCESS.getCode(), "success", result);
    }
}
